package com.example.mq.mqserver.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MQException;
import com.example.mq.mqserver.core.MQueue;
import com.example.mq.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.annotations.Param;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 这个类给上层提供文件存储消息的相关操作
 */
@Slf4j
public class MessageFileManager {

    /**
     * 用来获取指定队列的 消息文件所在目录
     * @param queueName 队列名
     * @return          返回该队列存放消息的目录
     */
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    /**
     * 用来获取指定队列的 消息数据文件路径
     * @param queueName 队列名
     * @return          返回该队列消息数据文件的路径
     */
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.dat";
    }

    /**
     * 用来获取该队列的 消息统计文件路径
     * @param queueName 队列名
     * @return          返回该队列消息统计文件的路径
     */
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    /**
     * 这个静态内部类，用来表示存储在消息统计文件中的数据
     */
    static public class Stat {
        // 消息总数
        public int totalCount;
        // 有效消息数
        public int validCount;
    }

    /**
     * 读取指定队列的消息统计文件中的数据
     *
     * @param queueName 队列名
     * @return          消息统计数据
     */
    private Stat readStat(String queueName) {
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            // 这个文件中，存的是文本数据，所以可以通过 Scanner 来读取
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 将 Stat 写入指定队列的消息统计文件
     * @param queueName 队列名
     * @param stat      消息统计数据
     */
    private void writeStat(String queueName, Stat stat) {
        // OutputStream 打开文件，默认会清空之前的文件内容
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();    // 保证数据从缓冲区中刷新到文件中
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 根据队列名，创建队列存放消息的目录和文件
     * @param queueName 队列名
     */
    public void createQueueFiles(String queueName) throws IOException {
        // 1. 创建目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            // 目录不存在，创建
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("[MessageFileManager] 创建目录失败！baseDir=" + baseDir.getAbsolutePath());
            }
        }
        // 2. 创建消息数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("[MessageFileManager] 创建消息数据文件失败！queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }
        // 3. 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("[MessageFileManager] 创建消息统计文件失败！queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }
        // 4. 给消息统计文件，写入初始值 0 \t 0
        Stat stat = new Stat();
        stat.validCount = 0;
        stat.totalCount = 0;
        writeStat(queueName, stat);
    }

    /**
     * 根据队列名，删除队列存放消息的目录和文件
     * @param queueName 队列名
     */
    public void destroyQueueFiles(String queueName) throws IOException {
        // 1. 删除消息数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        // 2. 删除消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        // 3. 删除目录
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            throw new IOException("[MessageFileManager] 删除队列目录和文件失败！baseDir=" + baseDir.getAbsolutePath());
        }
    }

    /**
     * 检查指定队列的 目录和文件 是否存在
     * @param queueName 队列名
     */
    private boolean checkFilesExits(String queueName) {
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }
        return true;
    }

    /**
     * 将消息写入指定队列对应的消息文件中
     * @param queue     队列
     * @param message   消息
     */
    public void sendMessage(MQueue queue, Message message) throws MQException, IOException {
        // 1. 检查一些当前要写入的队列对应的文件是否存在
        if (!checkFilesExits(queue.getName())) {
            throw new MQException("[MessageFileManager] 队列对应的文件不存在！queueName=" + queue.getName());
        }
        // 2. 将 Message 序列化为 byte[]
        byte[] messageBinary = BinaryTool.toBytes(message);

        // 加锁，多个线程打开同一个文件进行写入，文件格式就错乱了
        synchronized (queue) {
            // 3. 计算出 messageBinary 在文件中存储的位置赋值给 offsetBeg 和 offsetEnd
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);

            // 4. 将 messageBinary 追加写入消息数据文件
            // append: true 追加写入
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // 1) 先写入消息长度
                    // 使用 dataOutputStream 可以写入 int 4个字节
                    dataOutputStream.writeInt(messageBinary.length);
                    // 2) 写入消息内容
                    dataOutputStream.write(messageBinary);
                }
            }

            // 5. 更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    /**
     * 将消息从指定的队列对应的消息文件中删除
     * 逻辑删除
     * @param queue     队列
     * @param message   消息
     */
    public void deleteMessage(MQueue queue, Message message) throws IOException, ClassNotFoundException {
        // 同样需要加锁，避免多个线程同时对同一个文件写入数据
        synchronized (queue) {
            // RandomAccessFile 支持随机访问，可以根据 message 的 offsetBeg 和 offsetEnd 找到文件中的数据
            //  "rw" 读写的方式打开文件
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 1. 先从文件读取对应的 Message 数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                // 调整光标位置到 message 内容开头位置
                randomAccessFile.seek(message.getOffsetBeg());
                // 读取数据，填充 bufferSrc 数组
                randomAccessFile.read(bufferSrc);
                // 2. 将 bufferSrc 反序列化为 Message
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                // 3. 将 isValid 设置为 0x0 （逻辑删除）
                diskMessage.setIsValid((byte) 0x0);
                // 4. 写回文件中
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                // 重新调整光标位置
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }

            // 5. 更新消息统计文件
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                // 正常来说不需要判断
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }
    }

    /**
     * 加载指定队列的消息文件中的消息
     *
     * @param queueName     队列名
     * @return              消息链表
     */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MQException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                // 记录当前光标位置
                int currentOffset = 0;
                // 循环读取消息
                while (true) {
                    // 1. 读取消息长度
                    int messageSize = dataInputStream.readInt();
                    // 2. 按照消息长度，读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if (messageSize != actualSize) {
                        throw new MQException("[MessageFileManager] 文件格式错误！queueName=" + queueName);
                    }
                    // 3. 反序列化
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    // 4. 判断该消息是否有效
                    if (message.getIsValid() == 0x1) {
                        // 使用当前光标位置计算 offsetBeg 和 offsetEnd
                        message.setOffsetBeg(currentOffset + 4);
                        message.setOffsetEnd(currentOffset + 4 + messageSize);
                        // 加入链表
                        messages.add(message);
                    }
                    // 5. 更新光标位置
                    currentOffset += 4 + messageSize;
                }
            } catch (EOFException e) {
                // 结束标识
                // dataInputStream.readInt() 读到文件末尾，会抛出 EOFException 异常
                log.info("恢复 Message 数据完成！queueName=" + queueName);
            }
        }
        return messages;
    }

    /**
     * 检查指定队列的消息文件是否需要垃圾回收
     * 约定垃圾回收触发条件：总消息数 > 2000 并且 有效消息数 低于 总消息数 的 50%
     */
    public boolean checkGC(String queueName) {
        Stat stat = readStat(queueName);
        if (stat.totalCount > 50 && (double) stat.validCount / (double) stat.totalCount < 0.5) {
            return true;
        }
        return false;
    }

    /**
     * 获取新消息文件的路径
     */
    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.dat";
    }


    /**
     * 垃圾回收指定队列的消息数据文件
     *
     * 使用复制算法，创建一个新文件
     * 将旧文件中的有效消息，复制到新文件中
     * 然后删除就文件，将新文件重命名
     * 最后更新消息统计文件
     */
    public void gc(MQueue queue, MemoryDataCenter memoryDataCenter) throws MQException, IOException, ClassNotFoundException {
        // 加锁，GC 过程中，不允许其他线程对文件进行操作
        synchronized (queue) {
            // 统计 GC 消耗的时间
            long gcBeg = System.currentTimeMillis();
            // 1. 创建一个新文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()) {
                // 出现这种情况是因为，上次 GC 时，程序崩溃
                throw new MQException("[MessageFileManager] gc 时发现该队列的 "
                        + "queue_data_new.dat 已经存在！queueName=" + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if (!ok) {
                throw new MQException("[MessageFileManager] 创建新数据文件失败！ queueDataNewFile="
                        + queueDataNewFile.getAbsolutePath());
            }
            // 2. 从旧文件中，读出所有有效的消息
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            // 3. 将有效消息写入新数据文件
            try (OutputStream outputStream = new FileOutputStream(queueDataNewFile)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        // 序列化
                        byte[] buffer = BinaryTool.toBytes(message);

                        // 记录消息在新文件中的位置
                        message.setOffsetBeg(queueDataNewFile.length() + 4);
                        message.setOffsetEnd(queueDataNewFile.length() + 4 + buffer.length);

                        // 写入消息传递
                        dataOutputStream.writeInt(buffer.length);
                        // 写入消息内容
                        dataOutputStream.write(buffer);
                    }
                }
            }
            // 4. 删除旧文件
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if (!ok) {
                throw new MQException("[MessageFileManager] 删除旧数据文件失败！queueDataOldFile="
                        + queueDataOldFile.getAbsolutePath());
            }
            // 5. 重命名 queue_data_new.dat => queue_data.dat
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if (!ok) {
                throw new MQException("[MessageFileManager] 文件重命名失败！queueDataNewFile="
                        + queueDataNewFile.getAbsolutePath() +", queueDataOldFile="
                        + queueDataOldFile.getAbsolutePath());
            }
            // 6. 更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(), stat);

            // GC后，消息在文件中的位置发生了改变，需要重新加载内存中的消息
            // 需要注意，已经推送出去的消息，但是还没删除的消息（手动应答），是不能放会回队列中的
            // 更新 总消息表中的消息和 队列中的消息（注意队列中的消息要进行筛选，没被推送出去的消息，才能放回队列中）

            // 总消息表
            ConcurrentHashMap<String, Message> messageMap = memoryDataCenter.getMessageMap();
            // 该队列的待确认消息表
            ConcurrentHashMap<String, Message> waitMessageMap = memoryDataCenter.getQueueMessageWaitAckMap().get(queue.getName());
            LinkedList<Message> list = new LinkedList<>();

            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);

                if (!waitMessageMap.containsKey(message.getMessageId())) {
                    // 去除待确认消息表中存在的消息
                    list.add(message);
                }
            }

            memoryDataCenter.getQueueMessageMap().put(queue.getName(), list);


            long gcEnd = System.currentTimeMillis();
            log.info("[MessageFileManager] gc 执行完毕！queueName"
                    + queue.getName() + ", time=" + (gcEnd - gcBeg) + "ms");
        }
    }
}
